package com.pxene.mr;

import java.io.IOException;

import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.lib.input.FileSplit;
import org.apache.hadoop.mapreduce.lib.output.MultipleOutputs;

import com.pxene.dmp.useragent.Parser;

public class CookieMap extends Mapper<LongWritable, Text, Text, Text> {
	private Text resKey = new Text();
	private Text resVal = new Text();
	private Parser parser = new Parser();
	private MultipleOutputs<Text, Text> mos;

	@Override
	protected void setup(Mapper<LongWritable, Text, Text, Text>.Context context) throws IOException, InterruptedException {
		mos = new MultipleOutputs<Text, Text>(context);
	}

	@Override
	protected void map(LongWritable key, Text value, Mapper<LongWritable, Text, Text, Text>.Context context) throws IOException, InterruptedException {
		String[] datas = null;
		FileSplit split = (FileSplit) context.getInputSplit();
		String path = split.getPath().toString();
		if (path.contains("dpiqixin") && path.contains("4G")) {
			datas = value.toString().split("\\|", 40);
			String url = datas[29].trim();
			String ua = datas[28].trim();
			if (ua.length() < 2)
				return;

			String device = parser.parseDevice(ua).family.trim();
			String os = parser.parseOS(ua).family.trim();
			
			if (("Other".equals(os) || "Other".equals(device))) {
				resKey.set(device + "|" + os);
				resVal.set(ua);
				mos.write(resKey, resVal, "error/");
			} else {
				resKey.set(device + "|" + os);
				resVal.set(ua);
				mos.write(resKey, resVal, "normal/");
			}

		} else if (path.contains("dpiqixin") && path.contains("3G")) {
			datas = value.toString().split("\\|", 38);
			String url = datas[27].trim();
			String ua = datas[26].trim();

			if (ua.length() < 2)
				return;

			String device = parser.parseDevice(ua).family.trim();
			String os = parser.parseOS(ua).family.trim();
			
			if (("Other".equals(os) || "Other".equals(device))) {
				resKey.set(device + "|" + os);
				resVal.set(ua);
				mos.write(resKey, resVal, "error/");
			} else {
				resKey.set(device + "|" + os);
				resVal.set(ua);
				mos.write(resKey, resVal, "normal/");
			}


		} else if (path.contains("dpifix")) {
			datas = value.toString().split("\\x05", 12);
			String url = datas[7].trim();
			String ua = datas[9].trim();

			if (ua.length() < 2)
				return;

			// String cookie = datas[10].trim();
			String device = parser.parseDevice(ua).family.trim();
			String os = parser.parseOS(ua).family.trim();
			
			if (("Other".equals(os) || "Other".equals(device))) {
				resKey.set(device + "|" + os);
				resVal.set(ua);
				mos.write(resKey, resVal, "error/");
			} else {
				resKey.set(device + "|" + os);
				resVal.set(ua);
				mos.write(resKey, resVal, "normal/");
			}

		}

	}

	@Override
	protected void cleanup(Mapper<LongWritable, Text, Text, Text>.Context context) throws IOException, InterruptedException {
		mos.close();
	}
}
